use std::{
    collections::{HashMap, HashSet},
    mem::MaybeUninit,
};

use common_base::{prelude::*, uuid::Uuid};

use super::{DatasetOwner, DatasetRef, RowsetOwner, RowsetRef};

/// Note: RowsetRef must be a self-reference. for safety reason, RowsetRef must not `pub(crate)`

// #[derive(Debug, serde::Serialize, serde::Deserialize)]
// #[serde(crate = "self::serde")]
// #[serde(rename_all = "camelCase")]
#[derive(Debug, Clone)]
pub struct Dataset {
    //
    pub(crate) uuid: String,
    pub(crate) dataset_type: DatasetType,
    pub(crate) table_name: String,
    pub(super) master: Option<RowsetRef>,
    pub(super) self_ref: MaybeUninit<DatasetRef>,
    //
    pub(super) rowset_list: Vec<RowsetRef>,
    rowset_owner_list: Vec<RowsetOwner>, // when dataset is drop, rowset will be drop automatic
    pub(super) parent_rowset: Option<RowsetRef>,
    pub(super) child_rowset: Vec<RowsetRef>,
    //
    pub(crate) dataset_pool: bool,
    pub(crate) start_data_pool: bool,
    pub(super) insert_data_pool: HashSet<RowsetRef>,
    pub(super) update_data_pool: Option<DatasetOwner>,
    pub(super) delete_data_pool: Option<DatasetOwner>,
    //
    pub(crate) prime_key: Vec<String>,
    pub(crate) prime_key_key_map: HashMap<String, HashMap<String, Vec<String>>>,
    pub(super) prime_key_rowset_map: HashMap<String, RowsetRef>,
    //
    pub(super) dataset_map: HashMap<String, DatasetOwner>,
    // parentTableName
    // filterDataSetMap
    // hierarchyDataSet
    // userObject
    //
}

// we must guarantee that: `master`/`parent_rowset`/`child_rowset` and other NonNull field must NOT used between threads!!!
unsafe impl Send for Dataset {}

impl Default for Dataset {
    fn default() -> Self {
        Self {
            uuid: Uuid::new_v4().to_string(),
            table_name: Default::default(),
            dataset_type: Default::default(),
            master: Default::default(),
            self_ref: MaybeUninit::uninit(),
            rowset_list: Default::default(),
            rowset_owner_list: Default::default(),
            parent_rowset: Default::default(),
            child_rowset: Default::default(),
            dataset_pool: Default::default(),
            start_data_pool: Default::default(),
            insert_data_pool: Default::default(),
            update_data_pool: Default::default(),
            delete_data_pool: Default::default(),
            prime_key: Default::default(),
            prime_key_key_map: Default::default(),
            prime_key_rowset_map: Default::default(),
            dataset_map: Default::default(),
        }
    }
}

/// constructor
impl Dataset {
    pub fn get_instance(table_name: &str, dataset_type: DatasetType) -> Self {
        let mut dataset = Dataset::default();
        dataset.dataset_type = dataset_type;
        dataset.table_name = table_name.to_owned();
        // dataset.insert_data_pool = Some(Self::get_data_pool_instance(table_name).into());
        dataset.update_data_pool = Some(Self::get_data_pool_instance(table_name).into());
        dataset.delete_data_pool = Some(Self::get_data_pool_instance(table_name).into());
        dataset
    }

    /// from get_dataset_by_key rename to
    pub fn create_dataset_view(&mut self, prime_key: &str, prime_key_val: &str) -> DatasetOwner {
        let mut view = DatasetOwner::get_instance(&self.table_name, DatasetType::View);
        view.set_prime_key(
            self.prime_key
                .iter()
                .map(|x| x.as_str())
                .collect::<Vec<&str>>(),
        );

        let list = self.get_prime_key_by_key(prime_key, prime_key_val).clone();
        for index in list {
            let rowset = self
                .get_rowset(index.as_str())
                .expect("rowset has prime key, but not exsit in rowset_list");
            view.insert_rowset_ref(rowset.clone());
        }

        view
    }

    fn get_data_pool_instance(table_name: &str) -> Dataset {
        let mut dataset = Dataset::default();
        dataset.dataset_type = DatasetType::Map;
        dataset.table_name = table_name.to_owned();
        dataset.dataset_pool = true;
        dataset.start_data_pool = false;
        dataset
    }
}

/// serizlize and deserizlize
impl Dataset {
    pub fn deserizlize_from_json(mut dataset_json: serde_json::Value) -> Result<Self> {
        let table_name =
            if let Some(table_name) = dataset_json.get("tableName").and_then(|v| v.as_str()) {
                table_name.to_owned()
            } else {
                return Err(ErrorCode::RequestInvalid(
                    "can not deserizlize tableName from Dataset",
                ));
            };

        let dataset_pool =
            if let Some(dataset_pool) = dataset_json.get("dataSetPool").and_then(|v| v.as_bool()) {
                dataset_pool
            } else {
                return Err(ErrorCode::RequestInvalid(
                    "can not deserizlize dataSetPool from Dataset",
                ));
            };

        let start_data_pool = if let Some(start_data_pool) =
            dataset_json.get("startDataPool").and_then(|v| v.as_bool())
        {
            start_data_pool
        } else {
            return Err(ErrorCode::RequestInvalid(
                "can not deserizlize startDataPool from Dataset",
            ));
        };

        let (rowset_list, rowset_owner_list) = if let Some(rowset_array) = dataset_json
            .get_mut("rowSetArray")
            .and_then(|v| v.as_array_mut())
        {
            let mut rowset_list: Vec<RowsetRef> = Vec::with_capacity(rowset_array.len());
            let mut rowset_owner_list = Vec::with_capacity(rowset_array.len());
            for rowset in rowset_array.iter_mut() {
                let rowset_owner = RowsetOwner::deserizlize_from_json(rowset.take())?;
                rowset_list.push((&rowset_owner).into());
                rowset_owner_list.push(rowset_owner);
            }
            (rowset_list, rowset_owner_list)
        } else {
            return Err(ErrorCode::RequestInvalid(
                "can not deserizlize rowSetArray from Dataset",
            ));
        };

        let dataset_map = if let Some(dataset_map_json) = dataset_json
            .get_mut("dataSetMap")
            .and_then(|v| v.as_object_mut())
        {
            let mut dataset_map = HashMap::with_capacity(dataset_map_json.len());
            for (dataset_key, dataset) in dataset_map_json.iter_mut() {
                dataset_map.insert(
                    dataset_key.clone(),
                    Dataset::deserizlize_from_json(dataset.take())?.into(),
                );
            }
            dataset_map
        } else {
            return Err(ErrorCode::RequestInvalid(
                "can not deserizlize dataSetMap from Dataset",
            ));
        };

        Ok(Self {
            uuid: Uuid::new_v4().to_string(),
            table_name,
            dataset_type: DatasetType::default(),
            master: None,
            self_ref: MaybeUninit::uninit(),
            rowset_list,
            rowset_owner_list,
            parent_rowset: None,
            child_rowset: Vec::default(),
            dataset_pool,
            start_data_pool,
            insert_data_pool: HashSet::default(),
            update_data_pool: None,
            delete_data_pool: None,
            prime_key: Vec::default(),
            prime_key_key_map: Default::default(),
            prime_key_rowset_map: Default::default(),
            dataset_map,
        })
    }
}

// common attribute
impl Dataset {
    pub fn uuid(&self) -> &str {
        &self.uuid
    }

    pub fn dataset_type(&self) -> DatasetType {
        self.dataset_type
    }

    pub fn table_anme(&self) -> &str {
        &self.table_name
    }

    pub fn set_master(&mut self, master: RowsetRef) {
        if self.master.is_none() {
            self.master = Some(master);
            self.set_child_master_key(master);
        }
    }

    fn self_ref(&self) -> DatasetRef {
        unsafe { self.self_ref.assume_init_read() }
    }

    pub fn set_self_ref(&mut self, self_ref: DatasetRef) {
        self.self_ref.write(self_ref);
    }

    fn set_child_master_key(&mut self, master: RowsetRef) {
        for child in self.rowset_list.iter_mut() {
            child.set_master(master)
        }
    }
}

// operate on Rowset
impl Dataset {
    pub fn get_rowset_list(&self) -> &Vec<RowsetRef> {
        &self.rowset_list
    }

    pub fn get_rowset_list_mut(&mut self) -> &mut Vec<RowsetRef> {
        &mut self.rowset_list
    }

    pub fn sort_rowset_list(&mut self, columns: Vec<&str>, acendings: Vec<bool>) {
        debug_assert!(
            columns.len() == acendings.len(),
            "columns has different len with acendings"
        );

        let column_acendings = columns.iter().zip(acendings).collect::<Vec<_>>();
        if column_acendings.len() == 0 {
            return;
        }

        let mut ordered_index = {
            let mut ordered_rowset = self.rowset_list.iter().enumerate().collect::<Vec<_>>();
            ordered_rowset.sort_by(|&(_, rowset_0), &(_, rowset_1)| {
                for col_acend in column_acendings.iter() {
                    let col_name = col_acend.0;
                    let is_asc = col_acend.1;
                    if col_name.trim().is_empty() {
                        let col_value_0 = rowset_0.to_string();
                        let col_value_1 = rowset_1.to_string();

                        match col_value_0.cmp(&col_value_1) {
                            std::cmp::Ordering::Equal => continue,
                            x @ _ => return if is_asc { x } else { x.reverse() },
                        }
                    } else {
                        let col_value_0 = rowset_0.get_object(col_name).unwrap();
                        let col_value_1 = rowset_1.get_object(col_name).unwrap();

                        match col_value_0.cmp(&col_value_1) {
                            std::cmp::Ordering::Equal => continue,
                            x @ _ => return if is_asc { x } else { x.reverse() },
                        }
                    }
                }

                std::cmp::Ordering::Equal
            });

            ordered_rowset
                .iter()
                .map(|(index, _)| *index)
                .collect::<Vec<_>>()
        };

        for i in 0..self.rowset_list.len() {
            if ordered_index[i] != i {
                let mut current_i = i;
                loop {
                    let target_i = ordered_index[current_i];
                    ordered_index[current_i] = current_i;
                    if ordered_index[target_i] == target_i {
                        break;
                    }
                    self.rowset_list.swap(current_i, target_i);
                    self.rowset_owner_list.swap(current_i, target_i);
                    current_i = target_i;
                }
            }
        }
    }

    pub fn insert_rowset(&mut self, mut rowset: RowsetOwner) -> RowsetRef {
        rowset.set_master(self.master.expect("master for dataset is none."));

        let index = self.rowset_list.len() as i32 + 1;
        self.insert_rowset_with_index(index, rowset, false)
    }

    pub fn insert_rowset_ref(&mut self, mut rowset: RowsetRef) {
        rowset.set_master(self.master.expect("master for dataset is none."));
        let index = self.rowset_list.len() as i32 + 1;
        self.insert_rowset_ref_with_index(index, rowset, false);
    }

    pub fn set_parent_rowset(&mut self, parent_rowset: RowsetRef) {
        self.parent_rowset = Some(parent_rowset);
    }

    pub fn insert_child_rowset(&mut self, child_rowset: RowsetRef) {
        if self
            .child_rowset
            .iter()
            .find(|x| **x == child_rowset)
            .is_none()
        {
            self.child_rowset.push(child_rowset);
        }
    }

    fn insert_rowset_with_index(
        &mut self,
        mut index: i32,
        rowset: RowsetOwner,
        unique: bool,
    ) -> RowsetRef {
        let mut rowset_ref: RowsetRef = (&rowset).into();

        // if unique and prime_key's value already exist, will not insert
        if unique && self.contains_prime_key_value(self.get_prime_key_value(&rowset_ref)) {
            return rowset_ref;
        }

        if index < self.rowset_list.len() as i32 {
            if index > 0 {
                index -= 1;
            }
            self.rowset_list.insert(index as usize, (&rowset).into());
            self.rowset_owner_list.insert(index as usize, rowset);
        } else {
            self.rowset_list.push((&rowset).into());
            self.rowset_owner_list.push(rowset);
        }

        self.fill_rowset_property(&mut rowset_ref);

        rowset_ref
    }

    fn insert_rowset_ref_with_index(
        &mut self,
        mut index: i32,
        mut rowset: RowsetRef,
        unique: bool,
    ) {
        // if unique and prime_key's value already exist, will not insert
        if unique && self.contains_prime_key_value(self.get_prime_key_value(&rowset)) {
            return;
        }

        if index < self.rowset_list.len() as i32 {
            if index > 0 {
                index -= 1;
            }
            self.rowset_list.insert(index as usize, rowset);
        } else {
            self.rowset_list.push(rowset);
        }

        self.fill_rowset_property(&mut rowset);
    }

    fn fill_rowset_property(&mut self, rowset_ref: &mut RowsetRef) {
        rowset_ref.set_master(self.master.expect("master for dataset is none."));
        rowset_ref.set_master_table_name(self.table_name.as_str());

        self.build_prime_key_index(&rowset_ref);

        rowset_ref.insert_dataset_manager(self.self_ref());
        rowset_ref.set_owner_dataset(self.self_ref());

        self.insert_data_to_insert_pool(rowset_ref);
    }

    fn insert_data_to_insert_pool(&mut self, insert_rowset: &RowsetRef) {
        if !self.dataset_pool || !self.start_data_pool {
            return;
        }
        self.insert_data_pool.insert(insert_rowset.clone());
    }
}

/// operate about `Index`
impl Dataset {
    pub fn set_prime_key(&mut self, prime_key: Vec<&str>) {
        self.prime_key = prime_key.iter().map(|&v| v.to_owned()).collect();
    }

    pub fn build_prime_key_index(&mut self, rowset: &RowsetRef) {
        if self.prime_key.is_empty() {
            return;
        }

        let prime_key_value = self.get_prime_key_value(rowset);
        if let Some(v) = prime_key_value {
            self.prime_key_rowset_map.insert(v, rowset.clone());
        }
        self.build_prime_key_key_index(rowset);
    }

    fn get_prime_key_by_key(&mut self, prime_key: &str, prime_key_val: &str) -> &Vec<String> {
        let map = self.get_prime_key_key_map(prime_key);

        unsafe {
            (&*map)
                .get(prime_key_val)
                .expect("when get_prime_key_key_map, empty vec has been set to map")
        }
    }

    /// get value of prime_key
    /// if prime_key is multipy, values join with `-`
    fn get_prime_key_value(&self, rowset: &RowsetRef) -> Option<String> {
        if self.prime_key.is_empty() {
            None
        } else {
            let mut value = String::new();
            for i in 0..self.prime_key.len() {
                if i != 0 {
                    value.push_str("-");
                }
                value.push_str(rowset.get_string(self.prime_key[i].as_str()));
            }
            Some(value)
        }
    }

    fn get_rowset(&self, key_value: &str) -> Option<&RowsetRef> {
        self.get_rowset_inner(key_value.split(",").collect())
    }

    fn get_rowset_inner(&self, key_value: Vec<&str>) -> Option<&RowsetRef> {
        let mut key = String::new();
        for i in 0..key_value.len() {
            if i > 0 {
                key.push('-');
            }

            if let Some(&str) = key_value.get(i) {
                key.push_str(str);
            } else {
                key.push_str("");
            }
        }

        self.prime_key_rowset_map.get(key.as_str())
    }

    /// check if the value of prime_key has exist
    fn contains_prime_key_value(&self, key: Option<String>) -> bool {
        if self.prime_key.is_empty() || key.is_none() || self.prime_key_rowset_map.is_empty() {
            false
        } else {
            self.prime_key_rowset_map
                .contains_key(key.unwrap().as_str())
        }
    }

    fn build_prime_key_key_index(&mut self, rowset: &RowsetRef) {
        unsafe {
            for i in 0..self.prime_key.len() {
                let prime_key_slice = &self.prime_key[i] as *const String;
                let map = self.get_prime_key_key_map(prime_key_slice.as_ref().unwrap());

                let key = rowset.get_string(prime_key_slice.as_ref().unwrap());
                let value = self.get_prime_key_value(rowset);
                if let Some(value) = value {
                    let map = map.as_mut().unwrap();
                    let list = if map.contains_key(key) {
                        map.get_mut(key).unwrap()
                    } else {
                        map.entry(key.to_owned()).or_insert(Vec::default())
                    };
                    if !list.contains(&value) {
                        list.push(value);
                    }
                } else {
                    continue;
                }
            }
        }
    }

    fn get_prime_key_key_map(&mut self, prime_key: &str) -> *mut HashMap<String, Vec<String>> {
        if self.prime_key_key_map.contains_key(prime_key) {
            self.prime_key_key_map.get_mut(prime_key).unwrap()
        } else {
            self.prime_key_key_map
                .entry(prime_key.to_owned())
                .or_insert(HashMap::default())
        }
    }
}

/// operate on DatsetMap
impl Dataset {
    pub fn get_dataset_map(&self) -> &HashMap<String, DatasetOwner> {
        &self.dataset_map
    }
}

#[derive(Debug, Clone, Copy)]
pub enum DatasetType {
    Map,
    View,
    Container,
}

impl Default for DatasetType {
    fn default() -> Self {
        DatasetType::Container
    }
}
